package net.flyinggroup.reactor;

import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.FluxSink;

import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;

@Slf4j
public class FluxGenerate {
    @Test
    public void testFibonacciFluxSink() {
        Flux<Long> fibonacciGenerator = Flux.create(sink -> {
            long current = 0, next = 1;
            AtomicBoolean stop = new AtomicBoolean(false);
            sink.onDispose(() -> {
                stop.set(true);
                log.info("******Stop Received********");
            });
            sink.onCancel(() -> {
                stop.set(true);
                log.info("******Cancel Received********");
            });
            sink.onRequest(nums -> {
                log.info("******onRequest Received********");
                log.info("num is : {}", nums);
            });
            while (current >= 0) {
                log.info("generated" + current);
                sink.next(current);
                long prev = current;
                current = next;
                next = prev+current;
            }
            sink.complete();
        }, FluxSink.OverflowStrategy.DROP);
        List<Long> fibonacciSeries = new LinkedList<>();
        fibonacciGenerator.take(5).subscribe(num -> {
            System.out.println("consuming:" + num);
            fibonacciSeries.add(num);
        });
        System.out.println(fibonacciSeries);
    }
}
